package com.canal.starter.client;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.canal.starter.transfer.TransponderFactory;
import com.canal.starter.config.CanalProperties;
import com.canal.starter.exception.CanalClientException;
import lombok.Setter;
import org.apache.commons.collections4.MapUtils;
import org.apache.commons.lang3.StringUtils;

import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.*;

/**
 * Abstract implements of the CanalClient interface
 * It help to initialize the canal connector and so on..
 *
 * @author chen.qian
 */
public abstract class AbstractCanalClient implements CanalClient {
    /**
     * running flag
     */
    @Setter
    private volatile boolean running;
    /**
     * customer config
     */
    private CanalProperties canalProperties;

    /**
     * TransponderFactory
     */
    protected final TransponderFactory factory;

    AbstractCanalClient(CanalProperties canalProperties, TransponderFactory factory) {
        Objects.requireNonNull(canalProperties, "canalConfig can not be null");
        Objects.requireNonNull(factory, "transponderFactory can not be null");
        this.canalProperties = canalProperties;
        this.factory = factory;
    }

    @Override
    public void start() {
        if (canalProperties == null && MapUtils.isEmpty(canalProperties.getInstances())) {
            throw new CanalClientException("can not get the configuration of canal client");
        }
        for (Map.Entry<String, CanalProperties.Instance> instanceEntry : canalProperties.getInstances().entrySet()) {
            CanalProperties.Instance instance = instanceEntry.getValue();
            CanalConnector connector;
            if (instance.isClusterEnabled()) {
                List<SocketAddress> addresses = new ArrayList<>();
                for (String zookeeperAddress : instance.getZookeeperAddress()) {
                    String[] entry = zookeeperAddress.split(":");
                    if (entry.length != 2) {
                        throw new CanalClientException("error parsing zookeeper address=" + zookeeperAddress);
                    }
                    addresses.add(new InetSocketAddress(entry[0], Integer.parseInt(entry[1])));
                }
                connector = CanalConnectors.newClusterConnector(addresses, instanceEntry.getKey(), instance.getUsername(), instance.getPassword());
            } else {
                connector = CanalConnectors.newSingleConnector(new InetSocketAddress(instance.getHost(), instance.getPort()), instanceEntry.getKey(), instance.getUsername(), instance.getPassword());
            }
            connector.connect();
            if (StringUtils.isNotBlank(instance.getFilter())) {
                connector.subscribe(instance.getFilter());
            } else {
                connector.subscribe();
            }
            connector.rollback();
            this.process(connector, instanceEntry);
        }
    }

    /**
     * To initialize the canal connector
     *
     * @param connector CanalConnector
     * @param config    config
     */
    protected abstract void process(CanalConnector connector, Map.Entry<String, CanalProperties.Instance> config);

    @Override
    public void stop() {
        // this.setRunning(false);
        this.running = false;
    }

    @Override
    public boolean isRunning() {
        return this.running;
    }
}
